package org.example.com.atguigu.day01;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

public class $01_RddCreate {
    /*
     * Spark Rdd 的创建方式有三种:
     * 1. 通过集合创建
     * 2. 通过读取文件创建
     * 3. 通过其他的RDD衍生
     * */

    // 通过集合创建
    @Test
    public void createRddByCollection() {
        // 1.创建JavaSparkContext对象
        SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("org.example.com.atguigu.day01.$01_RddCreate");
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 2. 读取集合数据创建RDD
        JavaRDD<Integer> parallelize = sc.parallelize(Arrays.asList(1, 2, 5, 12, 231, 12));
        // 3. 调用collect触发任务计算
        System.out.println(parallelize.collect());

    }

    // 通过读取文件创建
    @Test
    public void createRddByFile() {
        // 1.创建JavaSparkContext对象
        SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("org.example.com.atguigu.day01.$01_RddCreate");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> stringJavaRDD = sc.textFile("E:\\IdeaProject\\bigdata0228\\spark0228\\datas\\wc.txt");

        List<String> collect = stringJavaRDD.collect();

        System.out.println(collect);

    }

    // 通过其他的RDD衍生
    @Test
    public void createRddByRdd() {
        SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("org.example.com.atguigu.day01.$01_RddCreate");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> rdd1 = sc.textFile("E:\\IdeaProject\\bigdata0228\\spark0228\\datas\\wc.txt");

        JavaRDD<String> rdd2 = rdd1.flatMap(line -> {
            String arr[] = line.split(" ");
            return Arrays.asList(arr).iterator();
        });
        System.out.println(rdd2.collect());
    }

}
